﻿#include "WRWorkTcpCommClientRoutine.h"
#include "../WRWorkSwitchEvent.h"
#include "../WRWorkTcpCommConsts.h"
#include "../WRWorkTcpCommHelper.h"
#include "../WRWorkTcpCommPulsePacket.h"
#include "WRWorkTcpCommClientSession.h"

#include <QTimer>
#include <QCoreApplication>

//注册初始事件类型。
static WR32s g_eventType = QEvent::registerEventType();

//初始化例程。
WRWorkTcpCommClientRoutine::WRWorkTcpCommClientRoutine(WRWorkTcpCommDomain* owner)
{
    m_owner = owner;
    m_pulseTimer = 0;
    m_checkTimer = 0;
    m_ticks = 0;
    m_status = Closed;
    m_socket = 0;
    m_session = 0;

    m_rBufLen = WRWorkTcpComm::MaxRecvSize;
    m_rBufArr = new char[m_rBufLen];
    m_rDatLen = 0;

    m_wStatus = false;
    m_wDatLen = 0;
    m_wDatPkg = 0;
}
//清理当前对象所占用的相关资源。
WRWorkTcpCommClientRoutine::~WRWorkTcpCommClientRoutine(void)
{
    delete[] m_rBufArr;
}

//执行数据写稿操作。
void WRWorkTcpCommClientRoutine::doWritting(void)
{
    if (m_wDatPkg)
    {
        char* buffer;
        WR32s length = m_wDatPkg->buffer(&buffer);
        m_wDatLen += m_socket->write(buffer + m_wDatLen, length - m_wDatLen);
        if (m_wDatLen == length)
        {
            m_wDatPkg->fulfil(m_wDatLen);
            m_wDatPkg->release();
            m_wDatPkg = 0;
        }
    }
    else
    {
        m_wMutex.lock();
        if (m_wQueue.count() > 0)
        {
            m_wDatPkg = m_wQueue.dequeue();
            m_wMutex.unlock();

            char* buffer;
            WR32s length = m_wDatPkg->buffer(&buffer);
            m_wDatLen = m_socket->write(buffer, length);
            if (m_wDatLen == length)
            {
                m_wDatPkg->fulfil(m_wDatLen);
                m_wDatPkg->release();
                m_wDatPkg = 0;
            }
        }
        else
        {
            m_wStatus = false;
            m_wMutex.unlock();
        }
    }
}
//处理接收到的数据，返回余下的数据长度。
WR32s WRWorkTcpCommClientRoutine::process(const char* buf, WR32s len)
{
    while(len >= 8)
    {
        //寻找数据头
        WR32s idx = WRWorkTcpComm::checkHeader(buf, len);
        //未找到帧头，忽略所有数据
        if (idx < 0)
        {
            return 0;
        }
        //帧头非起始数据，数据出现丢失
        if (idx > 0)
        {
            buf += idx;
            len -= idx;
        }
        //数据长度不够，退出循化继续接收数据
        if (len < 8)
        {
            break;
        }
        //读取帧长
        WR32s l = WRBytes::read<WR32s>(buf + 4);
        //验证帧长度，以跳过假帧头
        if (l > WRWorkTcpComm::MaxDataSize)
        {
            buf += 4;
            len -= 4;
            continue;
        }
        //未接收到全部数据，退出循环并继续接收数据
        if (len < l)
        {
            break;
        }
        //验证帧尾部，如果不是帧尾，则表示前面的帧头为无效帧头，跳过
        if (WRBytes::read<WR32u>(buf + l - 4) != WRWorkTcpComm::FrameTailor)
        {
            buf += 4;
            len -= 4;
        }
        //验证版本，如果版本不正确，则表示前面的帧头是无效帧头，跳过
        if (WRBytes::read<WR32u>(buf + l - 8) != WRWorkTcpComm::FrameVersion)
        {
            buf +=4;
            len -= 4;
        }
        //进行数据帧处理，并将其后移
        procData(buf + 8, l - 16);
        buf += l;
        len -=l;
    }
    return len;
}

//执行接收到的数据帧处理。
void WRWorkTcpCommClientRoutine::procData(const char* buf, WR32s len)
{
    WR16u index = WRBytes::read<WR16u>(buf);
    buf += 2;
    len -= 2;
    if (index == 0xFFFF)//全局相关
    {
        WR16u cmdType = WRBytes::read<WR16u>(buf);
        buf += 2;
        len -= 2;
        switch (cmdType)
        {
        case WRWorkTcpComm::Start://打开反馈。
        {
            WR16u pkgIndex = WRBytes::read<WR16u>(buf);
            buf += 2;
            len -= 2;
            WR16u count = WRBytes::read<WR16u>(buf);
            buf += 2;
            len -= 2;
            WR64s ticks = WRBytes::read<WR64s>(buf);
            buf += 8;
            len -= 8;
            if (pkgIndex == 1)//首包
            {
                if (ticks == m_ticks)
                {
                    WRWorkTcpCommClientSession* session = new WRWorkTcpCommClientSession(this, m_session);
                    m_session->release();
                    m_session = session;
                    m_wMutex.lock();
                    m_status = Online;
                    m_wMutex.unlock();
                    m_owner->onChanged(WRWorkDomain::Online, m_session);
                }
                else
                {
                    const unsigned char* pLen =(const unsigned char*)buf;
                    const unsigned char* pDat = 0;
                    {
                        WR32s remain = count % 8;
                        if (remain == 0) pDat = (const unsigned char*)(buf + count);
                        else pDat = (const unsigned char*)(buf + count + 8 - remain);
                    }
                    for(WR32s i = 0; i < count; ++i)
                    {
                        WRType::fromBytes(pDat, *pLen);
                        pDat += (*pLen) * sizeof(QUuid);
                        ++pLen;
                    }
                }
            }
            else if (pkgIndex == 2)//第二包
            {
                if (m_session)
                {
                    m_session->release();
                    m_session = 0;
                }
                m_session = new WRWorkTcpCommClientSession(this, count, buf);
                m_ticks = ticks;
                m_wMutex.lock();
                m_status = Online;
                m_wMutex.unlock();
                m_owner->onChanged(WRWorkDomain::Online, m_session);
            }
        }
            break;
        default:
            break;
        }
    }
    else if (m_session)
    {
        m_session->procData(index, (const unsigned char*)buf, len);
    }
}

//当连接时调用此方法。
void WRWorkTcpCommClientRoutine::onConnect()
{
    if (m_status == Disconnected)
    {
        m_wMutex.lock();
        m_status = Connecting;
        m_wMutex.unlock();
        m_socket->connectToHost(m_owner->addr(), m_owner->port());
    }
}
//当出现错误后调用此方法。
void WRWorkTcpCommClientRoutine::onError(QAbstractSocket::SocketError)
{
    onDisconnected();
}
//当连接后调用此方法。
void WRWorkTcpCommClientRoutine::onConnected(void)
{
    m_available = 1;
    m_pulseTimer = startTimer(WRWorkTcpComm::PulseInterval);
    m_checkTimer = startTimer(WRWorkTcpComm::SessionTimeout);

    m_rDatLen = 0;
    WRDataPacket* packet = WRDataPacket::allocate<32>();
    {
        char * p;
        packet->buffer(&p);

        WRBytes::write((WR32u)WRWorkTcpComm::FrameHeader, p);//帧头
        WRBytes::write((WR32s)(32), p + 4);//帧长度
        WRBytes::write((WR16u)(0xFFFF), p + 8);//全局指令
        WRBytes::write((WR16u)WRWorkTcpComm::Start, p + 10);//类型（打开）
        WRBytes::write(m_ticks, p + 16);//时间戳
        WRBytes::write((WR32u)WRWorkTcpComm::FrameVersion, p + 24);//版本
        WRBytes::write((WR32u)WRWorkTcpComm::FrameTailor, p + 28);//帧尾
    }

    m_wMutex.lock();
    m_status = Initializing;
    m_wDatPkg = packet;
    m_wDatLen = 0;
    m_wStatus = true;
    m_wMutex.unlock();
    doWritting();
}
//当断开连接后调用此方法。
void WRWorkTcpCommClientRoutine::onDisconnected(void)
{
    if (m_pulseTimer != 0)
    {
        killTimer(m_pulseTimer);
        m_pulseTimer = 0;
    }
    if (m_checkTimer != 0)
    {
        killTimer(m_checkTimer);
        m_checkTimer = 0;
    }

    if (m_status == Online)
    {
        m_wMutex.lock();
        m_status = Disconnected;
        if (m_wDatPkg)
        {
            m_wDatPkg->fulfil(-1);
            m_wDatPkg->release();
        }
        while(m_wQueue.count())
        {
            m_wDatPkg = m_wQueue.dequeue();
            m_wDatPkg->fulfil(-1);
            m_wDatPkg->release();
        }
        m_wDatPkg = 0;
        m_wDatLen = 0;
        m_wStatus = false;
        m_wMutex.unlock();

        m_session->shutdown();
        m_owner->onChanged(WRWorkDomain::Offline);

        QTimer::singleShot(WRWorkTcpComm::TryConnInterval, this, SLOT(onConnect()));
    }
    else if (m_status != Closed)
    {
        m_wMutex.lock();
        m_status = Disconnected;
        m_wMutex.unlock();

        QTimer::singleShot(WRWorkTcpComm::TryConnInterval, this, SLOT(onConnect()));
    }

    m_socket->abort();
}
//当数据写入后调用此方法。
void WRWorkTcpCommClientRoutine::onWritten(qint64)
{
    doWritting();
}
//当数据到来后调用此方法。
void WRWorkTcpCommClientRoutine::onReadyRead(void)
{
    do
    {
        //读取数据。
        WR32s length = (WR32s)(m_socket->read(m_rBufArr + m_rDatLen, m_rBufLen - m_rDatLen));
        if (length <= 0)
        {
            break;
        }
        m_rDatLen += length;
        //处理数据
        length = process(m_rBufArr, m_rDatLen);
        if (length >0 && length != m_rDatLen)
        {
            memmove(m_rBufArr, m_rBufArr + m_rDatLen - length, length);
        }
        m_rDatLen = length;
    }while(true);

    ++m_available;
}

//处理定时器事件。
void WRWorkTcpCommClientRoutine::timerEvent(QTimerEvent *e)
{
    if (e->timerId() == m_pulseTimer)
    {
        bool wStatus = false;
        m_wMutex.lock();
        m_wQueue.append(WRWorkTcpComm::pulsePacket());
        wStatus = m_wStatus;
        m_wStatus = true;
        m_wMutex.unlock();
        if (!wStatus)
        {
            doWritting();
        }
    }
    else if (e->timerId() == m_checkTimer)
    {
        if (m_available == 0)
        {
            onDisconnected();
        }
        else
        {
            m_available = 0;
        }
    }
    else
    {
        QObject::timerEvent(e);
    }
}

//执行事件处理。
bool WRWorkTcpCommClientRoutine::event(QEvent *e)
{
    if (WRWorkSwitchEvent::identify(e))//开关处理
    {
        WRWorkSwitchEvent* e2 = static_cast<WRWorkSwitchEvent*>(e);
        if (e2->action() == WRWorkSwitchEvent::Start)
        {
            bool* result = e2->result<bool>();
            if (m_status == Closed)
            {
                m_socket = new QTcpSocket();
                m_socket->setReadBufferSize(WRWorkTcpComm::MaxDataSize);

                QObject::connect(m_socket, SIGNAL(connected()), this, SLOT(onConnected()));
                QObject::connect(m_socket, SIGNAL(disconnected()), this, SLOT(onDisconnected()));
                QObject::connect(m_socket, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
                QObject::connect(m_socket, SIGNAL(bytesWritten(qint64)), this, SLOT(onWritten(qint64)));
                QObject::connect(m_socket, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(onError(QAbstractSocket::SocketError)));

                m_wMutex.lock();
                m_status = Connecting;
                m_wMutex.unlock();

                m_owner->onChanged(WRWorkDomain::Offline);

                m_socket->connectToHost(m_owner->addr(), m_owner->port());
            }
            *result = true;
        }
        else if (e2->action() == WRWorkSwitchEvent::Close)
        {
            if (m_status != Closed)
            {
                m_wMutex.lock();
                m_status = Closed;
                if (m_wDatPkg)
                {
                    m_wDatPkg->fulfil(-1);
                    m_wDatPkg->release();
                }
                while(m_wQueue.count())
                {
                    m_wDatPkg = m_wQueue.dequeue();
                    m_wDatPkg->fulfil(-1);
                    m_wDatPkg->release();
                }
                m_wDatPkg = 0;
                m_wDatLen = 0;
                m_wStatus = false;
                m_wMutex.unlock();

                if (m_session)
                {
                    m_session->shutdown();
                }

                QObject::disconnect(m_socket, SIGNAL(connected()), this, SLOT(onConnected()));
                QObject::disconnect(m_socket, SIGNAL(disconnected()), this, SLOT(onDisconnected()));
                QObject::disconnect(m_socket, SIGNAL(readyRead()), this, SLOT(onReadyRead()));
                QObject::disconnect(m_socket, SIGNAL(bytesWritten(qint64)), this, SLOT(onWritten(qint64)));
                QObject::disconnect(m_socket, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(onError(QAbstractSocket::SocketError)));
                m_socket->abort();
                m_socket->deleteLater();
                m_socket = 0;

                if (m_pulseTimer != 0)
                {
                    killTimer(m_pulseTimer);
                    m_pulseTimer = 0;
                }
                if (m_checkTimer != 0)
                {
                    killTimer(m_checkTimer);
                    m_checkTimer = 0;
                }

                m_owner->onChanged(WRWorkDomain::Closed);
            }
        }
        e2->fulfil();
        return true;
    }
    else if ((WR32s)(e->type()) == g_eventType)//数据发送
    {
        doWritting();
        return true;
    }
    return QObject::event(e);
}

//执行写入操作。
bool WRWorkTcpCommClientRoutine::process(WRDataPacket* packet)
{
    QMutexLocker locker(&m_wMutex);
    if (m_status == Online)
    {
        m_wQueue.enqueue(packet);
        if (!m_wStatus)
        {
            m_wStatus = true;
            QCoreApplication::postEvent(this, new QEvent((QEvent::Type)g_eventType));
        }
        return true;
    }
    return false;
}
